package com.yanggu.flink.datastream_api.transform

import com.yanggu.flink.datastream_api.source.ClickSource
import org.apache.flink.streaming.api.scala._

/**
 * 使用reduce模拟sum, 使用reduce模拟max的功能
 */
object ReduceFunctionDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.addSource(new ClickSource)

    stream
      .map(data => (data.name, 1L))
      //DataStream => KeyedStream
      .keyBy(_._1)
      //KeyedStream => DataStream
      //实现求和的功能（模仿sum）
      .reduce((v1, v2) => (v1._1, v1._2 + v2._2))
      //这里返回true, 意味着所有的数据的key都相同, 被发到同一个slot上执行
      .keyBy(_ => true)
      //实现最大值的功能（模仿maxBy）
      .reduce((v1, v2) => (v1._1, v1._2.max(v2._2)))
      .print()

    env.execute()
  }

}

